package com.yc.mqtt.server;

import com.yc.mqtt.coder.ByteBufToWebSocketFrameEncoder;
import com.yc.mqtt.coder.WebSocketFrameToByteBufDecoder;
import com.yc.mqtt.enums.ProtocolEnum;
import com.yc.mqtt.handler.EmbedMqttServerHandler;
import com.yc.mqtt.utils.IpUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;

import javax.net.ssl.SSLContext;

public class MqttEmbedServer {

    private String PROTOCOL = "TLS";

    private SSLContext SERVER_CONTEXT;

    private static final String MQTT_CSV_LIST = "mqtt, mqttv3.1, mqttv3.1.1";


    private void intProtocolHandler(ChannelPipeline channelPipeline, ProtocolEnum protocolEnum) {
        //channelPipeline.addLast(new IdleStateHandler(0, 0, 3));
        switch (protocolEnum) {
            case MQTT:
                channelPipeline.addLast("encoder", MqttEncoder.INSTANCE);
                channelPipeline.addLast("decoder", new MqttDecoder());
                channelPipeline.addLast(new EmbedMqttServerHandler());
                break;
            case MQTT_WS_MQTT:
                channelPipeline.addLast("httpCode", new HttpServerCodec());
                channelPipeline.addLast("aggregator", new HttpObjectAggregator(5 * 1024 * 1024));
                channelPipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/", MQTT_CSV_LIST));
                channelPipeline.addLast("wsDecoder", new WebSocketFrameToByteBufDecoder());
                channelPipeline.addLast("wsEncoder", new ByteBufToWebSocketFrameEncoder());
                channelPipeline.addLast("decoder", new MqttDecoder());
                channelPipeline.addLast("encoder", MqttEncoder.INSTANCE);
                channelPipeline.addLast(new EmbedMqttServerHandler());
                break;
            case MQTT_WS_PAHO:
                channelPipeline.addLast("httpCode", new HttpServerCodec());
                channelPipeline.addLast("aggregator", new HttpObjectAggregator(5 * 1024 * 1024));
                channelPipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", MQTT_CSV_LIST));
                channelPipeline.addLast("wsDecoder", new WebSocketFrameToByteBufDecoder());
                channelPipeline.addLast("wsEncoder", new ByteBufToWebSocketFrameEncoder());
                channelPipeline.addLast("decoder", new MqttDecoder());
                channelPipeline.addLast("encoder", MqttEncoder.INSTANCE);
                channelPipeline.addLast(new EmbedMqttServerHandler());
                break;
        }
    }

    public static void main(String[] args) {
        MqttEmbedServer server = new MqttEmbedServer();
        server.start(8083);
    }

    public void start(int port) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap()
                    .group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel channel) throws Exception {
                            intProtocolHandler(channel.pipeline(), ProtocolEnum.MQTT_WS_PAHO);
                        }
                    })
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            // bind
            ChannelFuture future = bootstrap.bind(port).addListener((ChannelFutureListener) channelFuture -> {
                if (channelFuture.isSuccess()) {
                    System.out.println("服务端启动成功【" +
                            IpUtils.getHost() + ":" + port + "】");
                } else {
                    System.out.println("服务端启动失败【" + IpUtils.getHost() + ":" + port + "】");
                }
            });
            ;
            // wait util stop
            future.channel().closeFuture().sync();


        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // stop
//            try {
//                workerGroup.shutdownGracefully();
//                bossGroup.shutdownGracefully();
//            } catch (Exception e) {
//            }
        }

    }

}
